
#pragma once
#include <thread>
#include <regex>
#include <functional>
#include <condition_variable>

#include "nonstd/span.hpp"

#ifndef LOG_PRINT
#define LOG_PRINT  fmt::println
#define LOGD_PRINT fmt::println
#define LOGW_PRINT fmt::println
#define LOGE_PRINT fmt::println
#endif

//#include "ringbuffer.hpp"
#include "deque_ringbuf.h"

typedef std::vector<std::string> vstrings;
typedef std::vector<uint8_t>     varray;

typedef std::function<int(varray& data_buffer)> data_process_func;

typedef std::function<int()>  is_enabled_func;
typedef std::function<void()> notify_func;
typedef std::function<void(std::string&)> line_func;

static constexpr std::size_t N = 16 * 1024;
class SimpleC : public std::enable_shared_from_this<SimpleC>
{
protected:
    virtual void close() 
    {
    };

    virtual int  read(const nonstd::span<char>& arr, unsigned int timeout_ms)
    {
        return -1;
    }

    void start()
    {
        std::thread t1([&]() {
            int ret;
            while (1)
            {
                if ((ret = rb_handle()) < 0)
                {
                    LOGE_PRINT("Oops, I/O operation has been aborted. {}", ret);
                    break;
                }
            }
            exit_flag_.store(1);
        });
        rb_th = std::move(t1);
    }

public:
    ~SimpleC()
    {
        std::call_once(close_once_, [&]() {
            close();

            exit_flag_.store(1);
            rb_cv.notify_one();

#ifndef _MSC_VER
            rb_th.detach();
            std::this_thread::sleep_for(std::chrono::seconds(1));
#else
            if (rb_th.joinable())
                rb_th.join();
#endif
        });
    }

    virtual int write(const nonstd::span<char>& arr)
    {
        return -1;
    }

    void handle(data_process_func process_fn, int m_interval_ms = 500)
    {
        std::mutex m_mutex;

        varray data_buffer(N);
        data_buffer.clear();

        while (!exit_flag_.load())
        {
            if (rb.empty())
            {
                std::lock_guard<std::mutex> locker(m_mutex);

                std::cv_status status = rb_cv.wait_for(m_mutex, std::chrono::milliseconds(m_interval_ms));
                if (status == std::cv_status::timeout)
                {
                    if (data_buffer.begin() != data_buffer.end())
                    {
                        LOGW_PRINT("timeout discard: ({})", data_buffer.size());
                        data_buffer.clear();
                    }
                    continue;
                }
            }
            else
            {
                rb_read(data_buffer);

                int offset = process_fn(data_buffer);
                if (offset < 0)
                {
                    LOGE_PRINT("data process error. {}", offset);
                    break;
                }
                data_buffer.erase(data_buffer.begin(), data_buffer.begin() + offset);
            }
        }
    }

    void handle_lines(line_func line_fn, is_enabled_func is_enabled_fn, notify_func notify_fn, int m_interval_ms = 500)
    {
        handle(
            [&](varray& data_buffer) {
                process_lines(data_buffer, line_fn, is_enabled_fn, notify_fn);
                return 0;
            },
            m_interval_ms);
    }

private:
    void process_lines(varray& data_buffer, line_func line_fn, is_enabled_func is_enabled_fn, notify_func notify_fn)
    {
        bool notify = false;
re:

        auto it = std::find_if(data_buffer.begin(), data_buffer.end(), [](uint8_t c) {
            return c == (uint8_t)'\n';
        });

        if (it != data_buffer.end())
        {
            if ((it - data_buffer.begin()) != 1 || data_buffer.at(0) != '\r')
            {
                std::string s = std::string(data_buffer.begin(), it);
                if (is_enabled_fn())
                {
                    line_fn(s);

                    notify = true;
                }
                else
                {
                    LOGD_PRINT("    >> {}", s);
                }
                
            }
            data_buffer.erase(data_buffer.begin(), it + 1);

            if (!data_buffer.empty())
            {
                goto re;
            }
        }

        if (notify)
            notify_fn();
    }

    void rb_read(varray& data_buffer)
    {
        std::lock_guard<std::mutex> locker(rb_mutex);
        if ((data_buffer.size() + rb.size()) > N)
        {
            LOGW_PRINT("capacity discard: ({})", data_buffer.size());
            data_buffer.clear();
        }
        std::copy(rb.begin(), rb.end(), std::back_inserter(data_buffer));
        rb.erase(rb.begin(), rb.end());
    }

    void rb_write(const nonstd::span<char>& arr)
    {
        std::lock_guard<std::mutex> locker(rb_mutex);

        std::copy(arr.begin(), arr.begin() + arr.size(), std::back_inserter(rb));

        rb_cv.notify_one();
    }

    int rb_handle()
    {
        std::array<char, 1024> data;

        int len;
        do
        {
            if (exit_flag_.load())
                return -1;

            if ((len = (int)read(data, 100)) > 0)
            {
                rb_write(nonstd::span<char>{data.begin(), data.begin() + len});
            }
        } while (len == 0);

        return len;
    }

private:
    std::once_flag close_once_;

    std::atomic_int exit_flag_{0};

    baudvine::DequeRingBuf<uint8_t, N> rb;
    std::mutex                         rb_mutex;
    std::condition_variable_any        rb_cv;
    std::thread                        rb_th;
};

